package com.gt.stream

import jline.internal.InputStreamReader
import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Durations, StreamingContext}

import java.io.{BufferedReader, InputStream}
import java.net.{ServerSocket, Socket}

/**
 * 自定义数据源
 */
object Streaming_WC_DIY_02 {

  def main(args: Array[String]): Unit = {

    //1. 创建上下文对象
    val conf: SparkConf = new SparkConf().setAppName("xx").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Durations.seconds(1))

    val src: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver(9999))
    val result: DStream[(String, Int)] = src.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
    result.print()

    //6. 启动任务
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 自定义数据源[自定义数据采集器]
   * 实现监控指定端口
   */
  class MyReceiver(port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {
    private var flag = true

    //采集器启动执行此方法
    override def onStart(): Unit = {
      //启动采集器之后，启动一个单独的线程，用于接收 socketClient 发送的数据
      new Thread(new Runnable {
        override def run(): Unit = {
          //TODO 实现接收数据功能
          startReceiveMsg()
        }
      }).start()
    }

//    private def startReceiveMsg(): Unit = {
//      //1. 创建 socket server
//      val serverSocket: ServerSocket = new ServerSocket(port)
//      //2. 监控客户端，对每一个客户端启动一个线程-处理客户端发送过来的数据
//      while (true) {
//        val socket: Socket = serverSocket.accept()
//        val reader = new BufferedReader(new InputStreamReader(socket.getInputStream))
//        var msg = reader.readLine()
//        while (StringUtils.isNotBlank(msg)){
//          store(msg)
//          msg = reader.readLine()
//        }
//      }
//    }

    private def startReceiveMsg(): Unit = {
      //1. 创建 socket server
      val serverSocket: ServerSocket = new ServerSocket(port)
      //2. 监控客户端，对每一个客户端启动一个线程-处理客户端发送过来的数据
      while (flag) {
        val socket: Socket = serverSocket.accept()
        new Thread(new Runnable {
          override def run(): Unit = {
            val reader = new BufferedReader(new InputStreamReader(socket.getInputStream))
            var msg = reader.readLine()
            while (flag && StringUtils.isNotBlank(msg)) {
              store(msg)
              msg = reader.readLine()
            }
          }
        }).start()
      }
    }

    //关闭采集器执行方法
    override def onStop(): Unit = {
      flag = false
    }
  }

}
